package com.study.mq.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.support.ProducerListener;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * kafka admin
 *
 */
@Configuration
@Slf4j
public class KafkaAdminConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String kafkaBootstrapServers;

    @Value("${kafka.service.topic}")
    private String kafkaServiceTopic;

    /**
     * 定义一个KafkaAdmin的bean，可以自动检测集群中是否存在topic,不存在则创建
     */
    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        // 指定多个kafka集群多个地址，例如：192.168.2.11,9092,192.168.2.12:9092,192.168.2.13:9092
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServers);
        KafkaAdmin kafkaAdmin = new KafkaAdmin(configs);
        return kafkaAdmin;
    }


    @Bean
    public ProducerListener<?,?> producerListener() {
        return new ProducerListener<String, String>(){
            @Override
            public void onSuccess(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata) {
                log.info("{} 发送成功!", producerRecord.toString());
            }

            @Override
            public void onError(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata, Exception exception) {
                log.info("{} 发送失败!", producerRecord.toString());
                // TODO 重新发送? 还是做什么后续操作？
            }
        };
    }

    @Bean
    public KafkaAdminClient adminClient(KafkaAdmin kafkaAdmin) {
        return (KafkaAdminClient) KafkaAdminClient.create(kafkaAdmin.getConfigurationProperties());
    }

    /**
     * 创建 Topic
     */
    @Bean
    public NewTopic testTopic(KafkaAdminClient adminClient) {
        // 创建topic，需要指定创建的topic的"名称"、"分区数"、"副本数量(副本数数目设置要小于Broker数量)"
        NewTopic newTopic = new NewTopic(kafkaServiceTopic, 2, (short) 2);
        adminClient.createTopics(List.of(newTopic));
        return newTopic;
    }
}